package com.infinitus.nginxInfo.task;


import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import javax.annotation.Resource;
import javax.persistence.EntityManagerFactory;

import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.NameValuePair;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.infinitus.nginxInfo.config.FetchUrlConfig;
import com.infinitus.nginxInfo.config.KafkaConfig;
import com.infinitus.nginxInfo.config.RedisKeyConfig;
import com.infinitus.nginxInfo.config.StaticCache;
import com.infinitus.nginxInfo.service.ConfigInfoService;
import com.infinitus.nginxInfo.service.KafkaLogService;
import com.infinitus.nginxInfo.service.RedisService;

@Component
public class LoadDataTask {
	
	private static final Logger LOG = LoggerFactory.getLogger(LoadDataTask.class);
	
	@Resource(name = "redisServiceDB0") 
	private RedisService redisService;
	@Autowired
	private ConfigInfoService configInfoService;
	@Autowired
	private KafkaLogService kafkaLogService;
	@Autowired
	private KafkaConfig kafkaConfig;
	@Autowired
	private FetchUrlConfig fetchUrlConfig;
	
	@Autowired
	private EntityManagerFactory entityManagerFactory;
	
	@Scheduled(cron = "0 0 0/2 * * ?") 
	public void updateExpiryDate() {
		Calendar calendar = Calendar.getInstance();
		calendar.setTime(new Date());
		calendar.add(Calendar.DATE, -7);
		calendar.set(Calendar.HOUR_OF_DAY, 0);
		calendar.set(Calendar.MINUTE, 0);
		calendar.set(Calendar.SECOND, 0);
		calendar.set(Calendar.MILLISECOND, 0);
		FilterRules.expiryDate = calendar.getTime();
	}
	
	@Scheduled(cron = "0 0/5 * * * ?") 
	public void loadFieldTypes() {
		loadFieldType(kafkaConfig.getCleanNginxTopic());
		loadFieldType(kafkaConfig.getCleanSwitchTopic());
	}
	
	/**
	 * 加载各个字段的类型
	 */
	public void loadFieldType(String topic) {
		CloseableHttpClient httpclient = HttpClients.createDefault();
		String token = StaticCache.getTopicToken(topic);
		boolean succ = false;
		if(FilterRules.isNotBlank(token)){
			try{
				HttpGet httpGet = new HttpGet(fetchUrlConfig.getFiledTypeUrl(token));
				CloseableHttpResponse response = httpclient.execute(httpGet);
				try {
					if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
						HttpEntity entity = response.getEntity();
						if (entity != null) {
							String res = EntityUtils.toString(entity, "UTF-8");
							if(res!=null && !"".equals(res.trim())){
								JSONObject json = JSONObject.parseObject(res);
								JSONArray types = json.getJSONArray("result");
								if(types!=null && types.size()>0){
									for(int i =0; i<types.size(); i++){
										StaticCache.setFieldType(topic+types.getJSONObject(i).getString("name"), types.getJSONObject(i).getString("type"));
										redisService.set(RedisKeyConfig.FIELD_TYPE+topic+types.getJSONObject(i).getString("name"), types.getJSONObject(i).getString("type"));
									}
									succ = true;
								}
							}
						}
					}
				} finally {
					response.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			} finally {
				try {
					httpclient.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		
		if(!succ){
			Set<Serializable> keys = redisService.keys(RedisKeyConfig.FIELD_TYPE+topic+"*"); 
			for(Serializable s : keys ){
				String kk = (String)s;
				StaticCache.setFieldType(kk.substring(kk.indexOf(topic)), redisService.get(kk));
			}
		}
	}
	
	@Scheduled(cron = "0 0 15 * * ?") 
	public void loadTopicTokens() {
		LOG.info("[kafka数据清洗模块] 加载项目id与token对应关系，以token获取维度类型 ");
		Map<String, String> topicTokens = configInfoService.getTopicTokens();
		StaticCache.setTopicTokens(topicTokens);
	}
	
	public static <T> List<T>  shuguoUrlQuery(String sql, Class<T> clazz) {
		System.out.println("查询语句:"+sql);
		List<T> list = null;
		CloseableHttpClient httpclient = HttpClients.createDefault();
		try {
			String qurl = FetchUrlConfig.getShuguoUrl();
//			String qurl = "http://insight-uba.infinitus.com.cn/api/plyql/sql";
			if(qurl==null){
				return null;
			}
			HttpPost httppost = new HttpPost(qurl);
			List<NameValuePair> formparams = new ArrayList<NameValuePair>();
			formparams.add(new BasicNameValuePair("query",sql));
			UrlEncodedFormEntity uefEntity = new UrlEncodedFormEntity(formparams, "UTF-8");
			httppost.setEntity(uefEntity);
			long startTime = System.currentTimeMillis();
			CloseableHttpResponse response = httpclient.execute(httppost);
			try {
				if(response.getStatusLine().getStatusCode() == HttpStatus.SC_OK){
					HttpEntity entity = response.getEntity();
					if (entity != null) {
						String res = EntityUtils.toString(entity, "UTF-8");
						System.out.println(res);
						JSONObject resObj = JSONObject.parseObject(res);
						list = JSONObject.parseArray( resObj.getString("result"), clazz);
						System.out.println("查询耗时"+(System.currentTimeMillis() -startTime)/1000f+"秒");
					}
				}
			} finally {
				response.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				httpclient.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return list;
	}
	
	
	private static Map add(Map<String,Object> all , Map<String, Object> map) {
		Set<Entry<String, Object>> entrySet = map.entrySet();
		for(Entry<String, Object> e : entrySet){
			Object value = all.get(e.getKey());
			if(value!=null){
				if(value instanceof Map){
					all.put(e.getKey(), add(new HashMap<String,Object>(), (Map)value));
				}
			}else{
				all.put(e.getKey(), e.getValue());
			}
		}
		return all;
	}
	
	
	public static void main(String[] args) throws ClientProtocolException, IOException {
	}
}
